package com.yesun.task.pipeline.worker;

import com.yesun.task.pipeline.common.Constant;
import com.yesun.task.pipeline.domain.dto.TaskDto;
import com.yesun.task.pipeline.handler.Executor;
import com.yesun.task.pipeline.handler.Loader;
import com.yesun.task.pipeline.sink.ClickhouseSink;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

//并行Worker
@Slf4j
public class ParallelWorker {
    volatile boolean stop = false;
    //每个Worker一个线程池
    private static List<ExecutorService> listExecutorService = new ArrayList<>();
    private static List<LinkedBlockingQueue<TaskDto>> listQueue = new ArrayList();
    private static Map<String,Executor> executorMap = new HashMap<>();
    //统计JOB执行情况，保存TaskName和JobName
    public static List<String> LIST_JOBS = new ArrayList<>();
    public static Map<String, String> MAP_TASK = new HashMap();

    //默认4个worker
    private int workerNum = 4;
    public ParallelWorker() {
        this(4);
    }
    public ParallelWorker(int workerNum) {
        this.workerNum = workerNum;
    }

    public int getWorkerNumber() {
        return this.workerNum;
    }

    //注册
    public static void registerLoader(String name, Loader loader) {
        log.info("register loader:" + name + " - "+ loader);
        //执行loader，每次触发就执行
        loader.load(listQueue.get(Math.abs(name.hashCode() % listQueue.size())));
    }
    public void registerExecutor(String name, Executor executor) {
        //重复注册会覆盖，但还是做判断，如果存在就不注册
        if(!executorMap.containsKey(name)) {
            log.info("register executor:" + name + " - "+ executor);
            executorMap.put(name, executor);
        }
    }

    //启动Workers，n个线程
    public void startWorkers() {
        for(int i=0; i<this.workerNum; i++) {
            //定义WORKER NAME
            final String name = "ParallelWorker-" + i;
            //初始化队列
            final LinkedBlockingQueue<TaskDto> queue = new LinkedBlockingQueue<>();
            listQueue.add(queue);
            //初始化线程池、
            final ExecutorService executorService = Executors.newFixedThreadPool(4);
            listExecutorService.add(executorService);

            RunableWithName runableWithName = new RunableWithName() {
                @Override
                public String getName() {
                    return name;
                }
                @Override
                public void run() {
                    log.info("Worker staring...");
                    while (!stop) {
                        try {
                            final TaskDto task = queue.take();
                            log.info("queue take task:" + task.getName());
                            //为了统计，触发后才加入
                            if(!LIST_JOBS.contains(task.getJobName())) {
                                LIST_JOBS.add(task.getJobName());
                            }
                            //FIXED
                            //MAP_TASK.put(task.getName(), task.getJobName());//重复的会更新 FIXED BUG应该是在Handler中分发时就写入到MAP_TASK

                            if(task.getTag().equals(Constant.JOB_TAG_RT_DETAIL)) {
                                //实时任务直接启动线程，不走线程池
                                //取得任务，分发给对应的Executor处理
                                Executor executor = executorMap.get(task.getTag());
                                if(executor != null) {
                                    RunableWithName runableWithName = new RunableWithName(){
                                        @Override
                                        public void run() {
                                            log.info("execute task:" + task.getName() + " executor:" + executor.toString());
                                            executor.execute(task);
                                            log.info("execute task finished! " + task.getName());
                                        }
                                        @Override
                                        public String getName() {
                                            return task.getName();
                                        }
                                    };
                                    //直接启动
                                    new Thread(runableWithName).start();
                                }
                            } else {
                                //分发给executor执行
                                Future<String> future = executorService.submit(new CallableWithName<String>(){
                                    //传递一个name进去
                                    @Override
                                    public String getName() {
                                        return task.getName(); //需要传递JobName，因为一个JOB可能有N个task
                                    }
                                    @Override
                                    public String call() {
                                        //取得任务，分发给对应的Executor处理
                                        Executor executor = executorMap.get(task.getTag());
                                        if(executor != null) {
                                            log.info("execute task:" + task.getName() + " executor:" + executor.toString());
                                            executor.execute(task);
                                            log.info("execute task finished! " + task.getName());
                                            return "success";
                                        }
                                        return "fail";
                                    }
                                });
                                try {
                                    String result = future.get();
                                    //成功，写入
                                    log.info("task["+task.getName()+"] 执行完毕，返回结果：" + result);
                                    //从TASK_MAP清理
                                    MAP_TASK.remove(task.getName());
                                    log.info("remove task: " + task.getName() + " MAP_TASK size:" + MAP_TASK.size());
                                } catch (ExecutionException e) {
                                    e.printStackTrace();
                                }
                            }
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            };
            //启动
            Thread thread = new Thread(runableWithName);
            thread.setName(name);
            thread.start();
        }
    }

    public class Monitor extends Thread {
        @Override
        public void run() {
            log.info("Monitor staring...");
            while (true) {
                try {
                    Thread.sleep(30000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                int i=0;
                for(ExecutorService executorService : listExecutorService) {
                    ThreadPoolExecutor threadPoolExecutor = ((ThreadPoolExecutor) executorService);
                    int threadCount = threadPoolExecutor.getActiveCount();
                    long taskCount = threadPoolExecutor.getTaskCount();
                    long completeTaskCount = threadPoolExecutor.getCompletedTaskCount();
                    log.info("****************************");
                    log.info("******线程池序号：" + i);
                    log.info("******活动线程数：" + threadCount);
                    log.info("******全部任务数：" + taskCount);
                    log.info("******完成任务数：" + completeTaskCount);
                    log.info("****************************");
                    i++;
                }
            }
        }
    }
}
